package com.bigdata.flink.datastream;

import com.bigdata.flink.util.StreamEnvUtil;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class CustomRichFunctionTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamEnvUtil.getEnv();
        DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

        env.setParallelism(1);

        SingleOutputStreamOperator<String> map = socketSource.map(new CustomMap() {
            @Override
            public String map(String value) throws Exception {
                return super.map(value);
            }
        });

        map.print();
        env.execute();
    }


    // rich function，函数调用前后可以执行一次open、close方式来处理统一资源
    // 也可以获取执行上下文环境
    static class CustomMap extends RichMapFunction<String, String> {

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("CustomMap 初始化成功");
        }

        @Override
        public String map(String value) throws Exception {
            RuntimeContext runtimeContext = getRuntimeContext();
            System.out.println(runtimeContext.getJobId());
            return value.split("\\s")[0];
        }

        @Override
        public void close() throws Exception {
            System.out.println("CustomMap 执行结束");
        }

        @Override
        public RuntimeContext getRuntimeContext() {
            RuntimeContext runtimeContext = super.getRuntimeContext();
            JobID jobId = runtimeContext.getJobId();
            System.out.println(jobId);
            return runtimeContext;
        }
    }
}
